/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.reserve;


import java.util.Calendar;
import java.util.Collection;
import java.util.LinkedList;

import jp.aist.gtrc.plus.scheduler.ScheduleOrder;
import jp.aist.gtrc.plus.scheduler.ScheduleStarter;
import jp.aist.gtrc.plus.scheduler.SchedulerOptions;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeManager;
import jp.aist.gtrc.plus.scheduler.status.StatusManager;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

import org.apache.log4j.Logger;


public class ReserveManager {

	private final transient ReserveTable table;
	private final transient ReserveRecorder recorder;
	private final transient NodeAllocator nodeAllocator;
	private final transient NodeManager nodeMgr;
	private final transient ReserveNodeNormalizer normalizer;
	private final transient ReserveOperationManager opeMgr;
	private final transient ScheduleStarter starter;
	private final transient StatusManager statusMgr;
	private transient int maxExpired;
	private transient ReserveOperation currentOpe;
	protected static final transient Logger logger = Logger.getLogger(ReserveManager.class);

	public ReserveManager(NodeManager nodeMgr, ScheduleStarter starter,
			ReserveRecorder recorder, NodeAllocator nodeAllocator,
			SchedulerOptions options, StatusManager statusMgr) throws ReserveException {

		assert(nodeMgr != null);
		try {
			this.table = new ReserveTable();
			this.recorder = recorder;
			this.starter = starter;
			this.statusMgr = statusMgr;
			this.nodeAllocator = (nodeAllocator != null)
				? nodeAllocator : new ReserveNodeAllocator(options, nodeMgr, statusMgr);
			this.nodeMgr = nodeMgr;
			this.maxExpired = options.getMaxExpiredReserves();
			this.currentOpe = ReserveOperation.getIdleInstance();

			this.normalizer = makeReserveNodeNormalizer();
			this.opeMgr = makeReserveOperationManager(options);

			System.setSecurityManager(new ReserveSecurityManager());
		} catch (ReserveException e) {
			if (this.recorder != null) {
				recorder.finish();
			}
			throw e;
		}

	}

	protected ReserveNodeNormalizer makeReserveNodeNormalizer() throws ReserveException {

		return new ReserveNodeNormalizer(this);

	}

	protected ReserveOperationManager makeReserveOperationManager(SchedulerOptions options)
		throws ReserveException {

		return new ReserveOperationManager(this, normalizer, options);

	}

	public synchronized void recoverFromRecorder() throws ReserveException {

		logger.debug("recover reserves start");

		Collection<ReserveInfo> allRsvs = recorder.loadAll(nodeMgr);
		table.clear();
		for (ReserveInfo info : allRsvs) {
			logger.debug("Recover Rsv: " + info);
			table.put(info);
		}

		opeMgr.recoverFromRecorder();

		logger.debug("recover reserves finished");

	}

	protected final ReserveTable getReserveTable() {

		return table;

	}

	protected final ReserveRecorder getReserveRecorder() {

		return recorder;

	}

	protected final NodeAllocator getNodeAllocator() {

		return nodeAllocator;

	}

	protected final StatusManager getStatusManager() {

		return statusMgr;

	}

	protected final ReserveNodeNormalizer getReserveNodeNormalizer() {

		return normalizer;

	}

	public final NodeManager getNodeManager() {

		return nodeMgr;

	}

	final synchronized void startCurrentOperation(ReserveOperation ope) throws ReserveException {

		if (currentOpe.getOperationType() != ReserveOperationType.idle) {
			logger.debug("commit/abort not called!! " + ope);
			throw new ReserveException("waiting commit/abort now, cannot do any other operation");
		}
		logger.debug("start " + ope);
		currentOpe = ope;

	}

	protected final synchronized void finishCurrentOperation() {

		assert(currentOpe.getOperationType() != ReserveOperationType.idle);
		currentOpe = ReserveOperation.getIdleInstance();

	}

	public final synchronized ReserveOperation getCurrentOperation() {

		return currentOpe;

	}

	protected void addNewReserveInfo(ReserveInfo rsvInfo) throws ReserveException {

		table.put(rsvInfo);
		recorder.store(rsvInfo);

	}

	private boolean toBeRemoved(ReserveInfo info) {

		Calendar now = Calendar.getInstance();
		if (info.isExpired(now) == false) {
			logger.debug("reserve has not expired yet");
			return false;
		}
		Collection<ReserveInfo> expired = table.getNotAfterEndTime(now);
		if (expired.size() < maxExpired) {
			logger.debug("remain expired reseve, maxExpiredNum=" + maxExpired
					+ " > curExpiredNum=" + expired.size());
			return false;
		}

		int index = 0;
		for (ReserveInfo r : expired) {
			if (r.compareTo(info) == 0) {
				break;
			}
			index++;
			if (index >= maxExpired) {
				break;
			}
		}
		logger.debug("remain expired reseve, maxExpiredNum=" + maxExpired
				+ ", curExpiredIdx=" + index);
		return (index >= maxExpired);

	}

	protected void removeReserveInfo(ReserveInfo rsvInfo, boolean forceRemove) throws ReserveException {

		if (rsvInfo.getReserveNodeSet() != null) {
			rsvInfo.unregisterReservedNodesForever();
		}

		if (forceRemove || toBeRemoved(rsvInfo)) {
			table.remove(rsvInfo);
			recorder.remove(rsvInfo);
		}

	}

	public final synchronized void remove(ReserveInfo info) throws ReserveException {

		logger.debug("Try remove: " + info);

		startCurrentOperation(ReserveOperation.getRemoveInstance(info));
		removeReserveInfo(info, false);
		opeMgr.commit(info.getReserveId());

		logger.debug("removed: " + info);

	}

	protected ReserveInfo newReserveInfo(ReserveId rsvId, ReserveRequest rsvReq,
			ReserveNodeSet rsvNodeSet) throws ReserveException {

		return new ReserveInfo(rsvId, rsvReq, rsvNodeSet);

	}

	public final synchronized ReserveServer getReserveServer() {

		return opeMgr;

	}

	protected boolean waitingTooLong(ReserveOperation ope, ReserveInfo target) {

		Calendar opStart = ope.getOperationStartTime();
		Calendar now = Calendar.getInstance();
		Calendar rsvStart = target.getStartTime();
		final long WAIT_MAX_MS = 3 * 60 * 1000; // 3 min

		if ((now.getTimeInMillis() - opStart.getTimeInMillis()) >= WAIT_MAX_MS) {
			return true;
		}

		if (opStart.before(rsvStart)) {
			return now.after(rsvStart);
		} else {
			Calendar rsvEnd = target.getEndTime();
			return now.after(rsvEnd);
		}

	}

	public synchronized boolean scheduleWaitingCommitAbort() throws ReserveException {

		ReserveOperation ope = getCurrentOperation();
		if (ope.getOperationType() == ReserveOperationType.idle) {
			return false;
		}

		ReserveInfo info = ope.getTargetReserveInfo();
		if (info == null) {
			/*
			 * reserve failed before making ReserveInfo by RuntimeException.
			 * This part must not be called actually.
			 */
			logger.debug("target reserve is null: ope=" + ope.toString());
			finishCurrentOperation();
			return false;
		}

		if (waitingTooLong(ope, info) == false) {
			return false;
		}

		if (ope.isAbortable()) {
			opeMgr.abort(info.getReserveId());
		} else {
			// To cancel running reserve is not abortable
			// because some jobs has already killed...
			opeMgr.commit(info.getReserveId());
		}

		return true;

	}

	protected final void kickScheduler(ReserveInfo info) {

		if (starter != null) {
			logger.debug("kick scheduler to run scheduling");
			try {
				ReserveId id = (info != null) ? info.getReserveId() : null;
				starter.runSchedule(id);
			} catch (SchedulerException e) {
				// ignore to continue
				logger.debug("cannot start scheduling", e);
			}
		}

	}

	public synchronized Collection<ReserveInfo> getExpiredReserve() {

		return getExpiredReserve(Calendar.getInstance());

	}

	public synchronized Collection<ReserveInfo> getExpiredReserve(Calendar now) {

		LinkedList<ReserveInfo> list = new LinkedList<ReserveInfo>();
		for (ReserveInfo info : table.getAllById()) {
			if (info.isExpired(now)) {
				// Check time here, not check whether state is DONE.
				// If added DONE reserves only, CANCELED/ERROR reserves
				// remain forever...
				list.add(info);
			}
		}
		return list;

	}

	public final synchronized void finish() {

		if (recorder != null) {
			recorder.finish();
		}

	}

	public final synchronized void stopServer() throws ReserveException {

		if (starter != null) {
			ScheduleOrder order = ScheduleOrder.getStopOrderInstance();
			starter.runSchedule(order);
			kickScheduler(null);
		}

	}

	public synchronized ReserveInfo getReserveInfo(ReserveId rsvId) {

		return table.get(rsvId);

	}

}
